import os

from dotenv import load_dotenv
from kafka import KafkaProducer
import json

from dbse.mysql.db_mysql_insert import getDataBase
from utils.DateUtil import getTodayStr


def commonProduceMsg(profile,topic,msg):
    home_path = os.environ.get("PYTHONPATH").split(";")[0]
    print(f'{home_path}\.env.{profile}')
    load_dotenv(f'{home_path}\.env.{profile}')
    servers = os.getenv('KAFKA_SERVERS').split(",")
    producer = KafkaProducer(
        bootstrap_servers=servers,
        # value_serializer=lambda v: json.dumps(v).encode('utf-8')  # 序列化器  会在字符串外围包上一层双引号
    )
    producer.send(topic, msg.encode('utf-8'))

def produceDfcfMsg(stockCode,db):
    producer = KafkaProducer(
        # bootstrap_servers=['39.98.82.109:29092','39.98.82.109:39092'],  # Kafka 服务器地址
        bootstrap_servers=['192.168.0.111:9092'],
        value_serializer=lambda v: json.dumps(v).encode('utf-8')  # 序列化器
    )
    try:
        query_sql = f'select * from stock_all where stock_code = "{stockCode}"'
        cursor = db.cursor()
        cursor.execute(query_sql)
        dfcfStock = cursor.fetchone()
        msg = {'id':dfcfStock[0], 'stockCode': dfcfStock[1] , 'stockName':dfcfStock[3], 'jys' : dfcfStock[4].lower()}
        producer.send('dfcf_stock_topic', msg)
        # 等待所有消息发送完成
        producer.flush()
        print(f"====异常代码:stockCode:{stockCode},重放成功====")

    except Exception as e:
        print(f"发送消息时出错: {e}")
    finally:
        # 关闭生产者
        producer.close()

def produceMsg():
    # 创建生产者
    producer = KafkaProducer(
        # bootstrap_servers=['39.98.82.109:29092','39.98.82.109:39092'],  # Kafka 服务器地址
        bootstrap_servers=['192.168.0.110:9092'],
        value_serializer=lambda v: json.dumps(v).encode('utf-8')  # 序列化器
    )

    # 发送消息
    try:
        # 发送单条消息
        producer.send('python-topic', {'key': 'value', 'message': 'Hello Kafka!'})

        # 发送多条消息
        # for i in range(10):
        #     producer.send('python-topic', {'number': i, 'message': f'Message {i}'})

        # 等待所有消息发送完成
        producer.flush()
        print("消息发送成功")

    except Exception as e:
        print(f"发送消息时出错: {e}")

    finally:
        # 关闭生产者
        producer.close()

# commonProduceMsg('prod',"cookie_expire",getTodayStr())
